package com.gitee.xhs.keyedState.aggregatingState;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import static com.gitee.xhs.StateConstant.FLINK_CHECK_POINT_PATH;

/**
 * @author BigHead
 * @date 2022/6/30 14:21
 * @description AggregatingStateConstant
 */
public class AggregatingStateConstant {

    public static final String CUSTOM_AGGREGATING_STATE_DIR = "aggregatingState/";

    public static final String AGGREGATING_STATE_PATH = FLINK_CHECK_POINT_PATH + CUSTOM_AGGREGATING_STATE_DIR;
    public static final String AGGREGATING_STATE_UID = "aggregating-state-uid";
    public static final String AGGREGATING_STATE_NAME = "aggregating-state";
    public static final AggregatingStateDescriptor<Transaction, AggregateFunctionDao, AggregateFunctionDao> AGGREGATING_STATE_DESCRIPTOR = new AggregatingStateDescriptor<Transaction, AggregateFunctionDao, AggregateFunctionDao>(
            AGGREGATING_STATE_NAME,
            new CustomAggregatingFunction(),
            TypeInformation.of(AggregateFunctionDao.class)
    );


    private static class CustomAggregatingFunction implements AggregateFunction<Transaction, AggregateFunctionDao, AggregateFunctionDao> {

        @Override
        public AggregateFunctionDao createAccumulator() {
            return new AggregateFunctionDao();
        }

        @Override
        public AggregateFunctionDao add(Transaction value, AggregateFunctionDao accumulator) {
            accumulator.add(value);
            return accumulator;
        }

        @Override
        public AggregateFunctionDao getResult(AggregateFunctionDao accumulator) {
            return accumulator;
        }

        @Override
        public AggregateFunctionDao merge(AggregateFunctionDao a, AggregateFunctionDao b) {
            a.merge(b);
            return a;
        }
    }

}
